package com.dahantc;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.time.temporal.TemporalUnit;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

@Slf4j
public class TaskConsumer extends Thread {

    private String topic;
    private int interval;

    public TaskConsumer(String topic, int interval) {
        this.topic = topic;
        this.interval = interval;
    }

    @Override
    public void run() {
        setName("Thread-Consumer-" + topic);
        log.info("==> TaskConsumer [topic={},interval={}] is beginning...", topic, interval);
        Map<String, Object> configs = KafkaUtils.buildConfig(true);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, topic + "_group1");
        // 创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(configs);
        // 可以使用正则表达式批量订阅主题
        final List<String> topics = Arrays.asList(topic);
        // 消费者订阅主题或分区
        consumer.subscribe(topics, new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                partitions.forEach(tp -> {
                    log.info("<== Revoked partition ：" + tp.partition());
                });
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                partitions.forEach(tp -> {
                    log.info("<== Assign partition" + tp.partition());
                });
            }
        });
        for (;;) {
            long wtime = System.currentTimeMillis();
            log.info("------------------------------------------------------------ [c-{}]", wtime%1000);
            log.info("==> Consumer message from [{}] C-START.", topic);
            // 拉取订阅主题的消息
            final ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));
            // 获取topic主题的消息
            final Iterable<ConsumerRecord<String, String>> topicIterable = records.records(topic);
            // 遍历topic主题的消息
            topicIterable.forEach(record -> log.info("====> ConsumerMsg[{}]", record.value()));
            consumer.commitSync(Duration.ofSeconds(3));
            log.info("==> Consumer message from [{}] C-END.", topic);
            log.info("------------------------------------------------------------ [c-{}]\n", wtime%1000);
            try {
                sleep(interval * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }
}
